package com.gxl.butt.service.impl;

import com.alibaba.dubbo.config.annotation.Service;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import com.gxl.butt.DAO.*;
import com.gxl.butt.domain.*;
import com.gxl.butt.domainkafka.KafkaUtil;
import com.gxl.butt.domainkafka.Message;
import com.gxl.butt.dto.bo.Token;
import com.gxl.butt.service.IkafkaOperatorService;

import com.gxl.butt.dto.bo.Topic;
import com.gxl.butt.thread.ExecutorThreadPool;
import com.gxl.butt.utils.DateUtil;
import com.gxl.butt.utils.JSONUtils;

import com.gxl.butt.utils.PageFinder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.KafkaException;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.util.Date;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;

import static com.gxl.butt.value.ErrorMsg.*;

/**
 * Created by guoxiaolin on 2018/7/13.
 */
@Service(interfaceClass = IkafkaOperatorService.class,retries = 0)
@Component
public class KafkaOperatorServiceImpl implements IkafkaOperatorService {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;


    @Autowired
    private IbttokenDAO ibttokenDAO;
    @Autowired
    private IKtopicDAO iKtopicDAO;
    @Autowired
    private ICountUserDAO iCountUserDAO;
    @Autowired
    private IReportInDayDAO iReportInDayDAO;
    @Autowired
    private IReportInMouthDAO iReportInMouthDAO;
    @Autowired
    private IHardUserDAO iHardUserDAO;
    @Autowired
    private ISumMessageDAO iSumMessageDAO;
    @Autowired
    private ItopicStatusDAO itopicStatusDAO;
    @Autowired
    private IReportInDayOfTopicDAO iReportInDayOfTopicDAO;


    private Logger log = LoggerFactory.getLogger(KafkaOperatorServiceImpl.class);
    private Gson gson = new GsonBuilder().create();

    @Override
    @Transactional
    public JSONObject sendMes(final Topic topic){
        boolean flag  = iKtopicDAO.hasTopic(topic.getTopicname());
        if (!flag){
            log.info("+++ topic 未创建 +++");
            return JSONUtils.getRoot(1,TOPIC_NOT_EXIST,STR_IS_EMPTY);
        }

        if (itopicStatusDAO.getTopicStatusByTopicName(topic.getTopicname()).get(0).getTopicstatus() != 1){
            return JSONUtils.getRoot(2,TOPIC_STATUS_STOP,STR_IS_EMPTY);
        }

        //topic 消息数量 +1
        iKtopicDAO.updateTopicNumbers(topic.getTopicname(),topic.getOpenid());

        boolean xcboo = false;
        try {
            xcboo = ExecutorThreadPool.fixedThreadPool(new Callable<Boolean>() {
                @Override
                public Boolean call() throws Exception {
                    Message message = new Message();
                    message.setId(System.currentTimeMillis());
                    message.setMsg(topic.getData().toString());
                    message.setSendTime(new Date());
                    kafkaTemplate.send(topic.getTopicname(), gson.toJson(message));
                    kafkaTemplate.send(TOPIC_OF_COUNT_USER, gson.toJson(topic.getOpenid()));//发送到countUser统计消息队列中
                    kafkaTemplate.send(TOPIC_OF_REPORT, gson.toJson(topic.getOpenid()));//发送到报表消息队列中
                    log.info("+++++++++++++++++++++ send message +++++++++++++++++++++");
                    return true;
                }
            });
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }

        if (xcboo){

            //计算当日当前topic累计
            FKTopic fkTopic = iKtopicDAO.getTopicInfoByTopicName(topic.getTopicname());
            ReportInDayOfTopic ridt = iReportInDayOfTopicDAO.getReportInDayOfTopicByTopicId(fkTopic.getId());
            if (ridt == null){
                ridt = new ReportInDayOfTopic();
                ridt.setCreateTime(String.valueOf(new Date().getTime()));
                ridt.setGmtTime(String.valueOf(new Date().getTime()));
                ridt.setInday(String.valueOf((new Date().getTime()/1000/60/60/24)*100000));
                ridt.setMqSum(1L);
                ridt.setTopicid(fkTopic.getId());
                ridt.setTopicname(topic.getTopicname());
                iReportInDayOfTopicDAO.saveDO(ridt);
            }else{
                ridt.setMqSum(ridt.getMqSum() + 1L);
                ridt.setGmtTime(String.valueOf(new Date().getTime()));
            }

            //计算累计
            new Thread(new Runnable() {
                @Override
                public void run() {
                    List<SumOfMessage> sumOfMessagelist = iSumMessageDAO.getDO();
                    SumOfMessage sumOfMessage =  sumOfMessagelist.get(0);
                    sumOfMessage.setMsgSum(sumOfMessage.getMsgSum() + 1L);
                    sumOfMessage.setGmtTime(String.valueOf(new Date().getTime()));
                    sumOfMessage.setCreateTime(String.valueOf(new Date().getTime()));
                }
            }).start();
            return JSONUtils.getRoot(0,SUCCESS_MSG,STR_IS_EMPTY);
        }

        return JSONUtils.getRoot(-1,FAILD_MSG,STR_IS_EMPTY);

    }

    @Override
    @Transactional
    public JSONObject createTopic(Topic topic){
        /** throws ExecutionException, InterruptedException **/
        Long uid = iHardUserDAO.selectUserByOpenid(topic.getOpenid());
        String newtopicname = "t" + uid + "p" + topic.getTopicname();
        topic.setTopicname(newtopicname);

        boolean flag  = iKtopicDAO.hasTopic(topic.getTopicname());
        log.info("++++++++++++++  topic 是否存在 ++++++++++++++" + flag);
        if (flag){
            return JSONUtils.getRoot(1,TOPIC_EXIST,flag);
        }

        log.info("++++++++++++++  topic 开始创建 +++++++++++++");
        try {
            KafkaUtil.createKafaTopic("139.199.233.133:80",topic);
        }catch (KafkaException e){
            e.printStackTrace();
        }
        log.info("++++++++++++++  topic 创建完成 ++++++++++++++");

        FKTopic kTopic = new FKTopic();
        kTopic.setCeatetime(String.valueOf(new Date().getTime()));
        kTopic.setGmttime(String.valueOf(new Date().getTime()));
        kTopic.setPartitionsum(topic.getPartitionsum());
        kTopic.setReplicationsum(topic.getReplicationsum());
        kTopic.setRemark(topic.getRemark());
        kTopic.setTopicname(topic.getTopicname());
        kTopic.setOpenid(topic.getOpenid());
        kTopic.setNumbers(0L);

        log.info("topic >> " + kTopic.toString());
        Long id = iKtopicDAO.saveTopic(kTopic);

        TopicStatus topicStatus = new TopicStatus();
        topicStatus.setTopicstatus(1);
        topicStatus.setTopicid(id);
        topicStatus.setTopicname(newtopicname);
        Long tsid = itopicStatusDAO.saveTopicStatus(topicStatus);

        Token topicBean = new Token();
        topicBean.setId(id);


        return JSONUtils.getRoot(0,SUCCESS_MSG,1);

    }

    @Override
    @Transactional
    public JSONObject getTopicList(Integer pageNo,Integer rowCount,Long userid) {


        List<FKTopic> fkTopics = iKtopicDAO.getTopicListByUserId(userid);
        Integer pageSize = fkTopics.size() / rowCount;
        Integer ci = fkTopics.size() % rowCount == 0 ? 0 : 1;

        PageFinder pf = new PageFinder(pageNo, (pageSize + ci), rowCount, fkTopics);
        return  JSONUtils.getRoot(0,SUCCESS_MSG, pf.getData());

    }

    @Override
    @Transactional
    public JSONObject getTopicInfo(Long tid) {

        FKTopic fkTopic =  iKtopicDAO.getTopicInfoByTopicId(tid);

        return  JSONUtils.getRoot(0,SUCCESS_MSG,fkTopic);

    }

    @Override
    @Transactional
    public JSONObject getCountOfUser(Long userid) {

        List<CountUser> countUsers = iCountUserDAO.getCountUserByUserid(userid);

        if (countUsers!=null && countUsers.size()>0){
            return  JSONUtils.getRoot(0,SUCCESS_MSG,countUsers);
        }

        return  JSONUtils.getRoot(1,SUCCESS_MSG,STR_IS_EMPTY);

    }

    @Override
    @Transactional
    public JSONObject getReportInDay(Long userid) {

        List<ReportInDay> reportInDays = iReportInDayDAO.getList(userid,30L);

        if (reportInDays != null && reportInDays.size() > 0 ){
            return  JSONUtils.getRoot(0,SUCCESS_MSG,reportInDays);
        }
        return  JSONUtils.getRoot(1,SUCCESS_MSG,STR_IS_EMPTY);

    }

    @Override
    @Transactional
    public JSONObject getReportInMouth(Long userid) {

        List<ReportInMouth> reportInMouths = iReportInMouthDAO.getList(userid,12L);

        if (reportInMouths != null && reportInMouths.size() > 0 ){
            return  JSONUtils.getRoot(0,SUCCESS_MSG,reportInMouths);
        }
        return  JSONUtils.getRoot(1,SUCCESS_MSG,STR_IS_EMPTY);

    }

    @Override
    @Transactional
    public JSONObject getSumOfMessage() {

        SumOfMessage sumOfMessage = iSumMessageDAO.getDO().get(0);

        return  JSONUtils.getRoot(0, SUCCESS_MSG, sumOfMessage);

    }


    @Override
    @Transactional
    public JSONObject getTopicsbyopenid(Integer pageNo,Integer pageSize,String openid) {

        Long uid = iHardUserDAO.selectUserByOpenid(openid);
        String tppre = "t" + uid + "p" ;

        List<FKTopic> fkTopics = iKtopicDAO.getTopicListByOpenid(openid);

        if (fkTopics == null || fkTopics.size() <= 0) {
            JSONObject redjson = new JSONObject();
            redjson.put("tplist","");
            redjson.put("tppre",tppre);
            return JSONUtils.getRoot(0,SUCCESS_MSG,redjson);
        }
        PageFinder pf = new PageFinder(pageNo,pageSize, fkTopics.size(),  fkTopics);

        Object ddo = pf.getResult();
        log.info(JSONObject.toJSON(ddo).toString());

        JSONArray jsonArray = JSONArray.parseArray(JSONObject.toJSON(ddo).toString());
        JSONArray resdatas = new JSONArray();
        Integer sp = pf.getStartOfPage();
        for (int i = sp ;i<jsonArray.size();i++){
            String createtime = jsonArray.getJSONObject(i).getString("ceatetime");
            Long topicid = jsonArray.getJSONObject(i).getLong("id");
            jsonArray.getJSONObject(i).put("createtime", DateUtil.timeStamp2DateInSSS(createtime,"yyyy年MM月dd日HH时mm分"));
            jsonArray.getJSONObject(i).put("runstatus", itopicStatusDAO.getTopicStatusByTopicId(topicid).get(0).getTopicstatus() == 0 ? "停止" : "运行");
            resdatas.add(jsonArray.getJSONObject(i));
            if (resdatas.size() == pageSize){
                break;
            }
        }

        JSONObject redjson = new JSONObject();
        redjson.put("tplist",resdatas);
        redjson.put("tppre",tppre);
        return  JSONUtils.getRoot(0,SUCCESS_MSG,redjson);

    }

    @Override
    @Transactional
    public JSONObject updataRemark(String remark, String topicid, String openid) {

        FKTopic fkTopic = iKtopicDAO.getTopicByOpenIdAndTopicId(topicid,openid).get(0);
        fkTopic.setRemark(remark);

        return JSONUtils.getRoot(0,SUCCESS_MSG,"");
    }

    @Override
    @Transactional
    public JSONObject updateTopicStatus(Long topicid, Integer topicstatus) {
        TopicStatus topicStatus = itopicStatusDAO.getTopicStatusByTopicId(topicid).get(0);
        if (topicStatus.getTopicstatus() != topicstatus){
            topicStatus.setTopicstatus(topicstatus);
            return JSONUtils.getRoot(0,SUCCESS_MSG,"");
        }
        return JSONUtils.getRoot(1,FAILD_MSG,"");
    }

    @Override
    @Transactional
    public JSONObject getTopicReportInWeek(Long topicid) {

        List<ReportInDayOfTopic> list = iReportInDayOfTopicDAO.getTopicReportInWeek(topicid);

        if (list == null){
            JSONObject jsonObject = new JSONObject();
            jsonObject.put("categories", new String[]{"0", "0", "0", "0", "0", "0", "0"});
            jsonObject.put("data",  new Long[]{0L,0L,0L,0L,0L,0L,0L});
            return JSONUtils.getRoot(0, SUCCESS_MSG, jsonObject);
        }
        String[] categories = new String[7];
        Long[] data =new Long[7];
        int i = 6;
        int j = 6;
        for (ReportInDayOfTopic repp : list) {
            categories[i--] = DateUtil.timeStamp2DateInday(repp.getInday(),null);
            data[j--] = repp.getMqSum();
        }

        JSONObject jsonObject = new JSONObject();
        jsonObject.put("categories", categories);
        jsonObject.put("data", data);

        return JSONUtils.getRoot(0, SUCCESS_MSG, jsonObject);

    }
}
